package com.bd03.cmccmonitor.util

import com.bd03.cmccmonitor.dbSql.SQLHandler
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD

object CmccKpi {
  //统计当前一小时的订单量,结果入库
  /**
   * //当天的日期,当前的小时,(数量,订单数量,消费时长,消费的金额)
   * (dayHourMinutes._1, dayHourMinutes._2, (1, succCount, costTime, chargefee))
   *
   * @param baseRDD
   * @return
   */
  def generalKpi(baseRDD: RDD[(String, String, (Int, Int, Long, Int),String,String)])={
    //当前一小时的结果数据
    val value: RDD[((String, String), (Int, Int, Long, Int))] = baseRDD.map(t=>((t._1,t._2),t._3))
    val generalHourRes: RDD[((String, String), (Int, Int, Long, Int))] = value.reduceByKey((a, b)=>((a._1+b._1),(a._2+b._2),(a._3+b._3),(a._4+b._4)))

    //当天的统计数据
    val generalDayRes: RDD[(String, (Int, Int, Long, Int))] = baseRDD.map(t=>(t._1,t._3)).reduceByKey((a, b)=>((a._1+b._1),(a._2+b._2),(a._3+b._3),(a._4+b._4)))

    //将当天的统计结果和当前一小时的统计结果入库
    SQLHandler.generalDay2Mysql(generalDayRes)
    SQLHandler.generalHour2Mysql(generalHourRes)

  }

  def provinceKpi(PCODE2PNAME: Broadcast[Map[String, AnyRef]],baseRDD: RDD[(String, String, (Int, Int, Long, Int),String,String)])={
    //当天各省份的成功充值订单
    val dayProvinceSuccRes: RDD[((String, String), (Int, Int))] = baseRDD.map(t=>((t._1,t._4),(t._3._1,t._3._2))).reduceByKey((a, b)=>((a._1+b._1),(a._2+b._2)))

    //将当天各省份的总订单和成功充值订单入库
    SQLHandler.provicneDay2Mysql( PCODE2PNAME,dayProvinceSuccRes)
  }


  def perMinutesKpi(baseRDD: RDD[(String, String, (Int, Int, Long, Int), String, String)]): Unit = {

    val perMinutesResult = baseRDD.map(tp => ((tp._1, tp._2, tp._5), (tp._3._2, tp._3._4))).reduceByKey((a, b) => (a._1 + b._1, a._2 + b._2))
    SQLHandler.perMinutes2MySQL(perMinutesResult)
  }
}
